package org.cc.cloud;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.cc.cloud.entity.Order;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.DefaultJackson2JavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.rabbitmq.client.Channel;
//spring amqp
@Configuration
public class RabbitMqConfig {
	
//	@Bean
//	public Exchange Exchange() {
//		return new Exchange("test", true);
//	}
	
	//spring容器启动的时候会判断有没有当前队列，没有会自动创建。
	@Bean
	public Queue debugQueue() {
		return new Queue("testQueue", true);
	}
	
	@Bean
	public Binding b1() {
		return new Binding("testQueue", Binding.DestinationType.QUEUE, "test", "test_key", new HashMap<String,Object>());
	}
	//SimpleMessageListenerContainer 消费的方式分为两种
	// MessageListener（监听）和MessageListenerAdapter（适配器）
	//MessageListener（监听）方式
	@Bean
	public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory) {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
		//SimpleMessageListenerContainer 支持一次监听多个queue
		container.setQueueNames("testQueue","error.queue");
		//container.setQueueNames("debug.queue");
		container.setConnectionFactory(connectionFactory);
		container.setConsumerTagStrategy(queue->{//修改tag。方便运维人员理解当前queue是干啥的
			return "测试"+queue;
		});
		//setConsumerArguments和setConsumerTagStrategy类似
		//点击队列进去之后。comsumers下面可以看到设置的ConsumerArguments和ConsumerTagStrategy信息
		
		//对消息进行后置处理在setMessageListener之前
		container.setAfterReceivePostProcessors(msg -> {
			System.out.println("----------原始消息----------");
			System.out.println(msg.getMessageProperties());
			System.out.println(new String(msg.getBody()));
			msg.getMessageProperties().getHeaders().put("desc", "这是一条后置处理过的消息");
			
			return new Message(("sms["+new String(msg.getBody())+"]").getBytes(), msg.getMessageProperties());
		});
		// 消费的一种方式
		
		container.setMessageListener(new MessageListener() {
			public void onMessage(Message message) {
				//message.getMessageProperties().getConsumerQueue()可以动态获取属于哪个队列
				System.out.println("----------SpringBoot rece msg---------");
				if("testQueue".equals(message.getMessageProperties().getConsumerQueue())){
					System.out.println("----------这条消息属于testQueue---------");
				}
				System.out.println(new String(message.getBody()));
			}
		});
		return container;
	}
	//SimpleMessageListenerContainer 消费的方式分为两种
	// MessageListener（监听）和MessageListenerAdapter（适配器）
	//MessageListenerAdapter（适配器）方式
	@Bean
	public SimpleMessageListenerContainer messageListenerAdapter(ConnectionFactory connectionFactory) {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
		//SimpleMessageListenerContainer 支持一次监听多个queue
		container.setQueueNames("mail","sms","weixin");
		//container.setQueueNames("debug.queue");
		container.setConnectionFactory(connectionFactory);	
		container.setConsumerTagStrategy(queue->{//修改tag。方便运维人员理解当前queue是干啥的
			return "测试"+queue;
		});
		// 消费的第二种方式
		MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageHandler());
		//如果设置了converter。会按照converter后的值去MessageHandler找参数相同的方法
		
		//查看源代码会发现原本就有一个默认convverter:SimpleMessageConverter
		//adapter.setMessageConverter(new TextMessageConverter());
		
		//转成json
		//json  MessageHandler  里面同时要添加一个Map类型的输入参数.
		//如果是集合那就是ArrayList 类型的输入参数
		//Jackson2JsonMessageConverter 发送方必须把contenttype 设置成"application/json"才行
		//adapter.setMessageConverter(new Jackson2JsonMessageConverter());
		
		
		//转成Java类的方式一：生产者设置"__TypeId__", "org.cc.cloud.entity.Order"指定了类，MessageHandler就可以接受指定类的参数了	
		//具体看first-client下的org.cc.cloud.controller.RabbitmqController类
		
		//转成Java类的方式二：生产者只要把"__TypeId__"设置成Map的key就行了。建议使用此种方式
		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		Jackson2JsonMessageConverter  jsonconverter=new Jackson2JsonMessageConverter();
		Map<String,Class<?>>  map=new HashMap<>();
		map.put("order", Order.class);
		DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
		javaTypeMapper.setIdClassMapping(map);
		jsonconverter.setJavaTypeMapper(javaTypeMapper);
		
		//adapter.setMessageConverter(jsonconverter);
		
		//代理模式
		ContentTypeDelegatingMessageConverter  converter=new ContentTypeDelegatingMessageConverter();
		converter.addDelegate("text", new TextMessageConverter());
		converter.addDelegate("application", jsonconverter);
		adapter.setMessageConverter(converter);
		//为每个队列设置单独的处理方法
		Map<String, String> queueOrTagToMethodName = new HashMap<>();
		queueOrTagToMethodName.put("weixin", "weixinMessage");
		queueOrTagToMethodName.put("mail", "mailMessage");
		queueOrTagToMethodName.put("sms", "smsMessage");
		adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
		
		container.setMessageListener(adapter);
		return container;
	}
	//手动确认消息的处理
	@Bean
	public SimpleMessageListenerContainer messageListenerContainerMANUAL(ConnectionFactory connectionFactory) {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
		//SimpleMessageListenerContainer 支持一次监听多个queue
		container.setQueueNames("error.queue");
		container.setConnectionFactory(connectionFactory);
		// 消费的一种方式
		
		/**
		 * AcknowledgeMode.NONE 不管报错不报错都自动确认，等效于 autoAck=true
		 * 
		 * AcknowledgeMode.MANUAL 手动确认，等效于 autoAck=false
		 * 
		 * AcknowledgeMode.AUTO 根据方法的执行情况来决定是否是确认还是拒绝（是否重新入queue）
		 * 1: 如果消息成功被消费了，则自动确认
		 * 2： 
		 *    (1)当抛出AmqpRejectAndDontRequeueException异常的时候，则消息会被拒绝，且requeue=false
		 *    (2)当抛出ImmediateAcknowledgeAmqpException异常，则消息会被确认
		 *    (3)其他的异常，则消息会被拒绝，且requeue=true（如果此时只有一个消费者监听该队列，则有发生死循环的风险）
		 *       其中，requeue=true可以通过setDefaultRequeueRejected（默认为true）去设置
		 *    
		 *    要想了解更新详细的代码，则可以分析SimpleMessageListenerContainer.doReceiveAndExecute方法
		 *
		 */
		container.setAcknowledgeMode(AcknowledgeMode.MANUAL);//如果不设置就是默认消息自动确认
		container.setMessageListener(new ChannelAwareMessageListener() {
			public void onMessage(Message message, Channel channel) throws Exception {
				System.out.println("+++++++++++++++onMessage++++++++++++++");
				System.out.println(new String(message.getBody()));
				TimeUnit.SECONDS.sleep(10);
				if(message.getMessageProperties().getHeaders().get("error") == null) {
				   //  消息类型，是否批量
					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
					System.out.println("消息已确认");
				} else {
					channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
					System.out.println("消息拒绝");
				}
				
			}
		});
		return container;
	}
	
	

}
